package cn.jstars.datatocloud.modules.monitor.task;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.stream.Collectors;

import org.nutz.lang.Lang;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.google.common.collect.Maps;

import cn.jstars.datatocloud.common.cache.Cache;
import cn.jstars.datatocloud.common.consts.Consts.DBConst;
import cn.jstars.datatocloud.common.consts.Consts.DataType;
import cn.jstars.datatocloud.common.utils.DataUtil;
import cn.jstars.datatocloud.common.utils.DateUtil;
import cn.jstars.datatocloud.common.utils.PageUtil;
import cn.jstars.datatocloud.modules.base.MqService;
import cn.jstars.datatocloud.modules.monitor.entity.Toimachinestatus;
import cn.jstars.datatocloud.modules.monitor.service.MonitorService;

/**
 * 历史监控数据级联任务
 *
 * @author Elvish
 */
public class HistoryTask implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(HistoryTask.class);

    private SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
    private MqService mqService;
    private MonitorService monitorService;
    private String year;
    private List<String> allDays;
    private long historyTaskDayIntervalTime;

    public HistoryTask(MqService mqService, MonitorService monitorService, int year, List<String> allDays,
                       long historyTaskDayIntervalTime) {
        this.mqService = mqService;
        this.monitorService = monitorService;
        this.year = year + "";
        this.allDays = allDays;
        this.setHistoryTaskDayIntervalTime(historyTaskDayIntervalTime);
    }

    @Override
    public void run() {

        if (Lang.isEmpty(this.allDays)) {
            return;
        }

        String today = DateUtil.getToday();

        for (String day : allDays) {
            try {
                if (sdf.parse(day).after(sdf.parse(today))) {
                    break;
                }

                // 今天之前的数据都算历史数据
                // 查询数据
                // 为了防止内存溢出，每次查询1W条
                List<Toimachinestatus> dataList = monitorService.findAllWithoutUpload(year, day);

                // 当天没有数据就级联下一天
                if (Lang.isEmpty(dataList)) {
                    break;
                }

                // 根据machineNo分组
                Map<String, List<Toimachinestatus>> groupMap = dataList.stream()
                    .collect(Collectors.groupingBy(Toimachinestatus::getMachineno));

                // 替换机台编码前缀
                replaceMachinePrefix(groupMap);

                // 根据分组循环切分集合大小并发送且更新数据库状态
                for (Entry<String, List<Toimachinestatus>> entry : groupMap.entrySet()) {

                    List<Toimachinestatus> list = entry.getValue();

                    // 切分集合大小，100
                    List<List<Toimachinestatus>> pageList = PageUtil.paging(list, 100);

                    for (List<Toimachinestatus> tmsList : pageList) {

                        try {
                            sendMessage(tmsList, entry.getKey(), day);
                        } catch (Exception e) {
                            break;
                        }
                        monitorService.updateUploadStatus(tmsList, year, day);

                    }

                }

                // 设置间隔时间，防止数据库崩溃,默认1分钟
                // Thread.sleep(historyTaskDayIntervalTime);

            } catch (ParseException e) {
                LOGGER.error("日志转换错误", e);
            } catch (Exception e) {
                LOGGER.error("历史数据级联错误", e);
            }

        }

    }

    /**
     * 替换机台编码前缀
     *
     * @param groupMap
     */
    private void replaceMachinePrefix(Map<String, List<Toimachinestatus>> groupMap) {

        for (Entry<String, List<Toimachinestatus>> entry : groupMap.entrySet()) {
            // 获取机台编码
            String machineNo = entry.getKey();

            for (Toimachinestatus toimachinestatus : entry.getValue()) {
                // 机器编码需要替换
                toimachinestatus.setMachineno(Cache.factoryName + "-" + machineNo);
            }
        }

    }

    /**
     * 发送消息
     *
     * @param dataList  数据集合
     * @param machineNo 原工厂机台编码
     * @param day       具体日期
     * @throws Exception
     */
    private void sendMessage(List<Toimachinestatus> dataList, String machineNo, String day) throws Exception {
        String toiMachineStatusModelGroups = JSONObject.toJSONString(dataList);

        Map<String, String> toiMachineStatusMap = Maps.newHashMap();
        toiMachineStatusMap.put("ToiMachineStatus", toiMachineStatusModelGroups);

        Map<String, Object> dataMap = Maps.newHashMap();
        dataMap.put("MachineNo", machineNo);
        dataMap.put("ToiMachineStatusModelGroups", toiMachineStatusMap);

        // 封装数据
        String sendMsg = DataUtil.getMessage(DataType.MONITOR_DATA, DBConst.COUNTON_PREFIX + year,
            "toimachinestatus_" + day, Cache.factoryId, dataMap);

        // 发送数据至云端
        mqService.convertAndSend(Cache.cloudMQQueueName, sendMsg);
    }

    public long getHistoryTaskDayIntervalTime() {
        return historyTaskDayIntervalTime;
    }

    public void setHistoryTaskDayIntervalTime(long historyTaskDayIntervalTime) {
        this.historyTaskDayIntervalTime = historyTaskDayIntervalTime;
    }
}
